package com.parsechina.hpush.broker.internal;

import com.parsechina.hpush.broker.parser.MessageBuilder;
import com.parsechina.hpush.broker.parser.PingRespMessage;
import com.parsechina.hpush.broker.spi.ProtocolProcessor;
import com.parsechina.hpush.broker.utils.NettyUtils;
import com.parsechina.hpush.broker.utils.StringUtils;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.CorruptedFrameException;
import io.netty.handler.codec.mqtt.MqttMessageType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import static io.netty.channel.ChannelFutureListener.CLOSE_ON_FAILURE;

/**
 * @author linfeng
 * @since 2016/11/22
 */
@ChannelHandler.Sharable
public class NettyMqttHandler extends ChannelInboundHandlerAdapter {

    private static final Logger LOG = LoggerFactory.getLogger(NettyMqttHandler.class);
    private final ProtocolProcessor protocolProcessor;

    public NettyMqttHandler(ProtocolProcessor processor) {
        protocolProcessor = processor;
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

        MessageBuilder messageBuilder = new MessageBuilder(msg);

        try {

            MqttMessageType messageType = messageBuilder.commonMessage().getMessageType();
            LOG.info("Received a message of type {}", messageType);

            switch (messageType) {
                case CONNECT:
                    protocolProcessor.processConnect(ctx.channel(), messageBuilder.connectMessage());
                    break;
                case SUBSCRIBE:
                    protocolProcessor.processSubscribe(ctx.channel(), messageBuilder.subscribeMessage());
                    break;
                case UNSUBSCRIBE:
                    protocolProcessor.processUnsubscribe(ctx.channel(), messageBuilder.unsubscribeMessage());
                    break;
                case PUBLISH:
                    protocolProcessor.processPublish(ctx.channel(), messageBuilder.publishMessage());
                    break;
                case PUBREC:
                    protocolProcessor.processPubRec(ctx.channel(), messageBuilder.pubRecMessage());
                    break;
                case PUBCOMP:
                    protocolProcessor.processPubComp(ctx.channel(), messageBuilder.pubCompMessage());
                    break;
                case PUBREL:
                    protocolProcessor.processPubRel(ctx.channel(), messageBuilder.pubRelMessage());
                    break;
                case DISCONNECT:
                    protocolProcessor.processDisconnect(ctx.channel(), messageBuilder.disconnectMessage());
                    break;
                case PUBACK:
                    protocolProcessor.processPubAck(ctx.channel(), messageBuilder.pubAckMessage());
                    break;
                case PINGREQ:
                    PingRespMessage pingResp = new PingRespMessage();
                    ctx.writeAndFlush(pingResp.build()).addListener(CLOSE_ON_FAILURE);
                    break;
                default:
                    break;
            }
        } catch (Exception ex) {
            ex.printStackTrace();
            LOG.error("Bad error in processing the message", ex);
            ctx.fireExceptionCaught(ex);
        }

    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        String clientId = NettyUtils.clientId(ctx.channel());
        if (StringUtils.isNotBlank(clientId)) {
            //if the channel was of a correctly connected client, inform messaging
            //else it was of a not completed CONNECT message or sessionStolen
            boolean stolen = false;
            Boolean stolenAttr = NettyUtils.sessionStolen(ctx.channel());
            if (stolenAttr != null && stolenAttr.equals(Boolean.TRUE)) {
                stolen = true;
            }
            protocolProcessor.processConnectionLost(ctx.channel(), clientId, stolen);
        }
        ctx.close();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        if (cause instanceof CorruptedFrameException) {
            LOG.warn("Error decoding a packet, probably a bad formatted packet, message: {}", cause.getMessage());
        } else {
            LOG.error("Error on networking", cause);
        }
        ctx.close();
    }

    @Override
    public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
        if (ctx.channel().isWritable()) {
            protocolProcessor.notifyChannelWritable(ctx.channel());
        }
        ctx.fireChannelWritabilityChanged();
    }

    //注释掉完成连接和断开连接异步处理事件
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
        if (evt instanceof MqttProtocolProcessor.ForceDisconnectEvent) {
            MqttProtocolProcessor.ForceDisconnectEvent disconnectEvent = (MqttProtocolProcessor.ForceDisconnectEvent) evt;
            protocolProcessor.forceCloseConnection(ctx.channel(), disconnectEvent.getNewChannel(), disconnectEvent.getMsg());
        } else if (evt instanceof MqttProtocolProcessor.CompleteConnect) {
            MqttProtocolProcessor.CompleteConnect completeConnectEvent = (MqttProtocolProcessor.CompleteConnect) evt;
            protocolProcessor.completeConnect(ctx.channel(), completeConnectEvent.getMsg());
        }
    }


}
